package com.gmail.dengtao.joe.group.transport.impl.handler;

import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.gmail.dengtao.joe.commons.util.ByteUtils;
import com.gmail.dengtao.joe.group.channel.Channel;
import com.gmail.dengtao.joe.group.exception.MemberNotReachableException;
import com.gmail.dengtao.joe.group.member.Member;
import com.gmail.dengtao.joe.group.member.MemberUtils;
import com.gmail.dengtao.joe.group.protocol.impl.DATAProtocol;
import com.gmail.dengtao.joe.group.protocol.impl.HELLOProtocol;
import com.gmail.dengtao.joe.group.transport.impl.AbstractSenderHandler;
import com.gmail.dengtao.joe.group.transport.impl.filter.DataFilter;
import com.gmail.dengtao.joe.transport.session.IdleStatus;
import com.gmail.dengtao.joe.transport.session.Session;
import com.gmail.dengtao.joe.transport.session.impl.DatagramSession;
import com.gmail.dengtao.joe.transport.session.impl.SocketSession;

/**
 * Default channel sender handler
 * @author <a href="mailto:joe.dengtao@gmail.com">DengTao</a>
 * @version 1.0
 * @since 1.0
 */
public class ChannelSenderHandler extends AbstractSenderHandler {

	private Logger LOGGER = LoggerFactory.getLogger(this.getClass());

	private Member local;

	private List<byte[]> helloList = new ArrayList<byte[]>();
	private Map<byte[], Object> dataList = new ConcurrentHashMap<byte[], Object>();
	private Map<Session, Member> memberMap = new ConcurrentHashMap<Session, Member>();
	
	public ChannelSenderHandler(Channel channel, Member local) {
		this.channel = channel;
		this.local = local;
	}

	@Override
	public void sessionCreated(Session session) throws Exception {
		// set idle time for write
		session.setIdleTime(IdleStatus.WRITE_IDLE, idleTimeForWrite);
		
		// Basic information
		InetSocketAddress remoteAddr = null;
		if (session instanceof SocketSession) {
			remoteAddr = (InetSocketAddress) ((SocketSession) session).getRemoteSocketAddress();
		} else if (session instanceof DatagramSession) {
			remoteAddr = (InetSocketAddress) ((DatagramSession) session).getRemoteSocketAddress();
		}
		Member remote = new Member(remoteAddr.getAddress().getHostAddress(), remoteAddr.getPort());
		Member member = channel.getMember(MemberUtils.getUniqueId(local, remote));
		if (member == null) {
			if (LOGGER.isDebugEnabled()) {
				LOGGER.error("[SENDER][CREATE][DISCONNECT][MEMBER:ILLEGAL][SESSION:" + session + "]");
			}
			session.close();
		} else {
			member.setLocalUuid(MemberUtils.getUniqueId(local, remote));
			member.setRemoteUuid(MemberUtils.getUniqueId(member, local));
			memberMap.put(session, member);
		}
	}

	@Override
	public void sessionOpened(Session session) throws Exception {
		Member member = memberMap.get(session);
		if (member == null) {
			if (LOGGER.isDebugEnabled()) {
				LOGGER.error("[SENDER][OPEN][DISCONNECT][MEMBER:UNKOWN][SESSION:" + session + "]");
			}
			session.close();
			return;
		} else {
			InetSocketAddress localAddr = null;
			if (session instanceof SocketSession) {
				localAddr = (InetSocketAddress) ((SocketSession) session).getLocalSocketAddress();
			} else if (session instanceof DatagramSession) {
				localAddr = (InetSocketAddress) ((DatagramSession) session).getLocalSocketAddress();
			}
			
			Member remote = channel.getMember(member.getLocalUuid());
			if (remote == null) {
				if (LOGGER.isDebugEnabled()) {
					LOGGER.debug("[SENDER][OPEN][DISCONNECT][MEMBER:ILLEGAL][LOCAL:/" + localAddr.getAddress().getHostAddress() + ":" + localAddr.getPort() + "][REMOTE:/" + member.getHost() + ":" + member.getPort() + "][SESSION:" + session + "]");
				}
				session.close();
				return;
			} else {
				remote.setReceiver(session);
				remote.addEvent(Member.Event.RECEIVER_CONNECTED);
				memberMap.put(session, remote);
				if (LOGGER.isDebugEnabled()) {
					LOGGER.debug("[SENDER][OPEN][LOCAL:/" + localAddr.getAddress().getHostAddress() + ":" + localAddr.getPort() + "][REMOTE:/" + remote.getHost() + ":" + remote.getPort() + "][MEMBER:" + remote + "]");
				}
				// SEND HELLO!!!
				// host/port/localUuid
				String data = local.getHost() + "/" + local.getPort() + "/" + MemberUtils.getUniqueId(local, remote);
				helloList.add(send(remote, HELLOProtocol.SYMBOL, data));
				remote.addEvent(Member.Event.RECEIVER_SEND_HELLO);
				if (LOGGER.isDebugEnabled()) {
					LOGGER.debug("[SENDER][HELLO][LOCAL:/" + localAddr.getAddress().getHostAddress() + ":" + localAddr.getPort() + "][REMOTE:/" + remote.getHost() + ":" + remote.getPort() + "][MEMBER:" + remote + "]");
				}
			}
		}
	}

	@Override
	public void sessionIdle(Session session, IdleStatus status)
			throws Exception {
		Member member = memberMap.get(session);
		if (member == null) {
			if (LOGGER.isDebugEnabled()) {
				LOGGER.error("[SENDER][IDLE][DISCONNECT][MEMBER:UNKOWN][IDLE_STATUS:" + status + "][SESSION:" + session + "]");
			}
			session.close();
			return;
		} else {
			if (member.hasStatus(Member.Status.AVAILABLE)) {
				if (LOGGER.isDebugEnabled()) {
					LOGGER.debug("[SENDER][IDLE][IDLE_STATUS:" + status + "][MEMBER:" + member + "]");
				}
				// Fire Member Idle!!!!!!!!!!
				interceptorStack.fireMemberIdle(channel, member, com.gmail.dengtao.joe.group.member.IdleStatus.WRITE_IDLE);
			} else if (member.hasStatus(Member.Status.RECEIVER_OK)) {
				if (LOGGER.isDebugEnabled()) {
					LOGGER.debug("[SENDER][IDLE][WAITE SENDER...][RECEIVER OK][IDLE_STATUS:" + status + "][MEMBER:" + member + "]");
				}
			} else {
				if (LOGGER.isDebugEnabled()) {
					LOGGER.debug("[SENDER][IDLE][CONNECTION IN PROGRESS...][RECEIVER NOT OK][IDLE_STATUS:" + status + "][MEMBER:" + member + "]");
				}
			}
		}
	}

	@Override
	public void dataNotSent(Session session, Object data) throws Exception {
		Member member = memberMap.get(session);
		if (member == null) {
			if (LOGGER.isDebugEnabled()) {
				LOGGER.error("[SENDER][NOTSENT][DISCONNECT][MEMBER:UNKOWN][DATA:" + data + "][SESSION:" + session + "]");
			}
			session.close();
			return;
		} else {
			if (member.hasStatus(Member.Status.AVAILABLE)) {
				if (LOGGER.isDebugEnabled()) {
					LOGGER.error("[SENDER][NOTSENT][DATA:" + data + "][MEMBER:" + member + "]");
				}
				if (dataList.containsKey(data)) {
					// Fire Data Not sent!!!!!!!!!
					interceptorStack.fireDataNotSent(channel, member, dataList.get(data));
				} else {
					if (LOGGER.isDebugEnabled()) {
						LOGGER.error("[SENDER][NOTSENT][UNKOWN DATA][DATA:" + data + "][MEMBER:" + member + "]");
					}
				}
			} else {
				if (LOGGER.isDebugEnabled()) {
					LOGGER.error("[SENDER][NOTSENT][DISCONNECT][MEMBER:ILLEGAL][DATA:" + data + "][MEMBER:" + member + "]");
				}
				session.close();
			}
		}
	}

	@Override
	public void dataSent(Session session, Object data) throws Exception {
		Member member = memberMap.get(session);
		if (member == null) {
			if (LOGGER.isDebugEnabled()) {
				LOGGER.error("[SENDER][SENT][DISCONNECT][MEMBER:UNKOWN][DATA:" + data + "][SESSION:" + session + "]");
			}
			session.close();
			return;
		} else {
			if (member.hasStatus(Member.Status.AVAILABLE)) {
				if (LOGGER.isDebugEnabled()) {
					LOGGER.debug("[SENDER][SENT][DATA:" + data + "][MEMBER:" + member + "]");
				}
				if (dataList.containsKey(data)) {
					// Fire Data Sent!!!
					interceptorStack.fireDataSent(channel, member, dataList.get(data));
				} else {
					if (LOGGER.isDebugEnabled()) {
						LOGGER.error("[SENDER][SENT][UNKOWN DATA][DATA:" + data + "][MEMBER:" + member + "]");
					}
				}
			} else if (member.hasStatus(Member.Status.RECEIVER_SEND_HELLO) && helloList.contains(data)) {
				helloList.remove(data);
				member.setReceiver(session);
				member.addEvent(Member.Event.RECEIVER_HELLO_SENT);
				if (member.hasStatus(Member.Status.AVAILABLE)) {
					if (LOGGER.isDebugEnabled()) {
						LOGGER.debug("[SENDER][SENT][MEMBER ADDED][MEMBER:" + member + "]");
					}
					// Fire Member Add!!!
					interceptorStack.fireMemberAdded(channel, member);
				} else {
					if (LOGGER.isDebugEnabled()) {
						LOGGER.debug("[SENDER][SENT][WAITE SENDER...][RECEIVER OK][MEMBER:" + member + "]");
					}
				}
			} else {
				if (LOGGER.isDebugEnabled()) {
					LOGGER.error("[SENDER][SENT][DISCONNECT][MEMBER:ILLEGAL][DATA:" + data + "][MEMBER:" + member + "]");
				}
				session.close();
			}
		}
	}

	@Override
	public void sessionClosed(Session session) throws Exception {
		Member member = memberMap.get(session);
		if (member == null) {
			if (LOGGER.isDebugEnabled()) {
				LOGGER.error("[SENDER][CLOSE][DISCONNECTED][MEMBER:UNKOWN][SESSION:" + session + "]");
			}
		} else {
			memberMap.remove(session);
			if (member.hasStatus(Member.Status.AVAILABLE)) {
				member.addEvent(Member.Event.RECEIVER_DISCONNECTED);
				if (LOGGER.isDebugEnabled()) {
					LOGGER.error("[SENDER][CLOSE][DISCONNECTED][MEMBER:" + member + "]");
				}
				// Close sender...
				if (member.getSender() != null) member.getSender().close();
				if (LOGGER.isDebugEnabled()) {
					LOGGER.debug("[SENDER][MEMBER DISAPPEARED][MEMBER:" + member + "]");
				}
				channel.removeMember(member);
				// Fire Member Disappeared!!!
				interceptorStack.fireMemberDisappeared(channel, member);
			} else {
				if (member.hasStatus(Member.Status.RECEIVER_CONNECTED) || member.hasStatus(Member.Status.RECEIVER_SEND_HELLO) || member.hasStatus(Member.Status.RECEIVER_OK)) {
					member.addEvent(Member.Event.RECEIVER_DISCONNECTED);
				}
				// Close sender...
				if (member.getSender() != null) member.getSender().close();
				if (LOGGER.isDebugEnabled()) {
					LOGGER.error("[SENDER][CLOSE][DISCONNECTED][MEMBER:" + member + "]");
				}
			}
		}
	}
	
	@Override
	public void exceptionCaught(Session session, Throwable cause) {
		Member member = memberMap.get(session);
		if (member == null) {
			if (LOGGER.isDebugEnabled()) {
				LOGGER.error("[SENDER][EXCEPTION][DISCONNECT][MEMBER:UNKOWN][SESSION:" + session + "]", cause);
			}
			session.close();
		} else {
			if (LOGGER.isDebugEnabled()) {
				if (member.hasStatus(Member.Status.AVAILABLE)) {
					LOGGER.warn("[SENDER][EXCEPTION][MEMBER:" + member + "]", cause);
				} else if (member.hasStatus(Member.Status.RECEIVER_OK)) {
					LOGGER.warn("[SENDER][EXCEPTION][RECEIVER OK][MEMBER:" + member + "]", cause);
				} else {
					LOGGER.warn("[SENDER][EXCEPTION][RECEIVER NOT OK][MEMBER:" + member + "]", cause);
				}
			}
			
			if (member.hasStatus(Member.Status.RECEIVER_NOT_CONNECT)) {
				channel.removeMember(member); 
				// Fire Exception Caught!!
				interceptorStack.fireExceptionCaught(channel, member, new MemberNotReachableException(cause));
			} else {
				// Fire Exception Caught!!
				interceptorStack.fireExceptionCaught(channel, member, cause);
			}
		}
	}
	
	public void send(Member target, Object data) {
		if (target.hasStatus(Member.Status.AVAILABLE)) {
			if (data instanceof byte[]) {
				dataList.put(send(target, DATAProtocol.SYMBOL, (byte[])data), data);
			} else {
				dataList.put(send(target, DATAProtocol.SYMBOL, data.toString().getBytes()), data);
			}
		} else {
			throw new MemberNotReachableException();
		}
	}
	
	private byte[] send(Member target, int type, String data) {
		return send(target, type, data.getBytes());
	}
	
	private byte[] send(Member target, int type, byte[] data) {
		if (!target.hasStatus(Member.Status.RECEIVER_NOT_CONNECT) || target.getReceiver() == null) {
			int length = 0;
			int pn = getPn();
			int sn = 0;
			byte mf = 0;
			//data = SixBitUtils.encode(data);
			
			length = ByteUtils.sizeOf(DataFilter.START_IDENTIFIER, length, type, pn, sn, mf, data, DataFilter.END_IDENTIFIER);
			ByteBuffer buffer = ByteBuffer.allocate(length);
			buffer.put(DataFilter.START_IDENTIFIER);	// Start Symbol
			buffer.putInt(length);						// Length
			buffer.putInt(type);						// Type
			buffer.putInt(pn);							// pn
			buffer.putInt(sn);							// sn
			buffer.put(mf);								// mf
			buffer.put(data);							// data
			buffer.put(DataFilter.END_IDENTIFIER);		// End Symbol
			buffer.flip();
			byte[] dst = new byte[length];
			buffer.get(dst);
			
			target.getReceiver().send(dst);
			
			return dst;
		} else {
			throw new MemberNotReachableException();
		}
	}

	private int pn = 0;
	
	private int getPn() {
		return pn ++;
	}
}